package com.atguigu.day04;

import com.atguigu.util.ClickEvent;
import com.atguigu.util.ClickSource;
import com.atguigu.util.UserViewCountPerWindow;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

public class Example2 {
    public static void main(String[] args) {
        var env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env
                .addSource(new ClickSource())
                .keyBy(r -> r.username)
                .window(TumblingProcessingTimeWindows.of(Time.seconds(10)))
                .aggregate(new CountAgg(), new WindowResult())
                .print();

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public static class CountAgg implements AggregateFunction<ClickEvent, Integer, Integer> {
        @Override
        public Integer createAccumulator() {
            return 0;
        }

        @Override
        public Integer add(ClickEvent value, Integer accumulator) {
            return accumulator + 1;
        }

        // 窗口关闭的时候调用
        @Override
        public Integer getResult(Integer accumulator) {
            return accumulator;
        }

        @Override
        public Integer merge(Integer a, Integer b) {
            return null;
        }
    }

    // 输入的泛型是Integer
    public static class WindowResult extends ProcessWindowFunction<Integer, UserViewCountPerWindow, String, TimeWindow> {
        @Override
        public void process(String key, Context ctx, Iterable<Integer> iterable, Collector<UserViewCountPerWindow> out) throws Exception {
            // Iterable<Integer> iterable只有一条数据！！！是getResult的返回值
            out.collect(new UserViewCountPerWindow(
                    key,
                    iterable.iterator().next(),
                    ctx.window().getStart(),
                    ctx.window().getEnd()
            ));
        }
    }
}
